package removeDanglingPage;

import java.io.IOException;

import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;

import mapred.AbstractMapper;
import mapred.util.Bytes;

public class RemoveDP_GetReady_Mapper extends AbstractMapper<BytesWritable, BytesWritable, BytesWritable, BytesWritable>{

	@Override
	public void map(BytesWritable key, BytesWritable value, OutputCollector<BytesWritable, BytesWritable> output,
			Reporter reporter) throws IOException {
		// TODO Auto-generated method stub
		int size = value.getLength();
		ReadBytes k = new ReadBytes(key);
		byte[] sp = new byte[16];
		sp = k.nextUrl();
		if(size > 0) {
			ReadBytes link = new ReadBytes(value);
			reporter.incrCounter("Remove", "map", 1);
			byte[] url0 = Bytes.concat(sp, new byte[] {0});
			output.collect(new BytesWritable(Bytes.concat(url0, Bytes.Integer2Bytes((size / 16)))),
					new BytesWritable(new byte[0]));
//			output.collect(new BytesWritable(Bytes.concat(sp, Bytes.Integer2Bytes((size / 16)))), 
//					new BytesWritable(new byte[0]));
			while(link.hasNextUrl()) {
				reporter.incrCounter("Remove", "fx", 1);
				byte [] url1 = Bytes.concat(link.nextUrl(), new byte[] {1});
				output.collect(new BytesWritable(Bytes.concat(url1, sp)), 
						new BytesWritable(new byte[0]));
//				output.collect(new BytesWritable(Bytes.concat(link.nextUrl(), sp)), 
//						new BytesWritable(new byte[0]));
			}
		}
//		String text = value.toString().trim();
//		if(!text.isEmpty()) {
//			String[] values = text.split("\t");
//			output.collect(new Text(key+" "+values.length), new Text(""));
//			for(int i = 0; i < values.length; i++) {
//				output.collect(new Text(values[i]+" "+key), new Text(""));
//			}
//		}
	}
}
